package com.flink.demo.jointable19;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Map;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * description
 *
 * @author zsyoung@qq.com
 * 2020/8/5 23:56
 */
public class WholeLoad extends RichMapFunction<String, Order> {
    private static final Logger LOGGER = LoggerFactory.getLogger(WholeLoad.class);

    ScheduledExecutorService executor = null;
    private Map<String, String> cache;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        executor.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try {
                    load();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }, 5, 5, TimeUnit.MINUTES);
    }

    private void load() throws Exception {
        Class.forName("com.mysql.jdbc.Driver");
        Connection conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/dim?characterEncoding=UTF-8", "root", "toor");
        PreparedStatement pst = conn.prepareStatement("select cityId,cityName from info");
        ResultSet resultSet = pst.executeQuery();
        while (resultSet.next()) {
            String cityId = resultSet.getString("cityId");
            String cityName = resultSet.getString("cityName");
            cache.put(cityId, cityName);
        }
        conn.close();
    }

    @Override
    public Order map(String value) throws Exception {
        JSONObject jsonObject = JSONObject.parseObject(value);
        Integer cityId = jsonObject.getInteger("cityId");
        String userName = jsonObject.getString("userName");
        String items = jsonObject.getString("items");
        String cityName = cache.get(cityId);
        return new Order(cityId, userName, items, cityName);
    }
}
